package com.datasync.system.controller;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.datasync.system.domain.SErrorLog;
import com.datasync.system.domain.SSyncConfig;
import com.datasync.system.service.ISErrorLogService;
import com.datasync.system.service.ISSyncConfigService;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.io.Serializable;
import java.sql.*;
import java.util.*;
import java.util.Date;
import java.util.stream.Collectors;

/**
 * mysql 同步ES
 */
@RestController
@RequestMapping("/system/mysqlToEs")
public class MysqlToEsController {

    private static final Logger log = LoggerFactory.getLogger(MysqlToEsController.class);

    //配置
    private static Map<Long,SSyncConfig> sSyncConfigMap = new HashMap<>();

    //表数据
    private Map<Object,String> tableNameMap = new HashMap();

    //多监听管理
    private static Map<Long,BinaryLogClient> mysqlMap = new HashMap<>();
    private static Map<Long,RestHighLevelClient> esMap = new HashMap<>();

    private static Map<Long,Connection> mMap = new HashMap<>();

    @Autowired
    private ISSyncConfigService isSyncConfigService;

    @Autowired
    private ISErrorLogService isErrorLogService;

    @RequestMapping("/start/{id}")
    public void start(@PathVariable("id") Long id){
        SSyncConfig sSyncConfig = isSyncConfigService.selectSSyncConfigById(id);
        sSyncConfigMap.put(id,sSyncConfig);
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                create(sSyncConfigMap.get(id));
            }
        });
        thread.start();
    }

    @RequestMapping("/end/{id}")
    public void end(@PathVariable("id") Long id) throws IOException {
        try{
            //移除binlog监听
            mysqlMap.get(id).disconnect();
            mysqlMap.remove(id);
            //移除ES连接
            esMap.get(id).close();
            esMap.remove(id);
            //移除配置连接
            sSyncConfigMap.remove(id);
            //移除MySQL连接
            mMap.remove(id);
        }catch (Exception e){
            log.error("任务关闭失败",e);
        }
    }

    /**
     * 数据重新同步
     */
    @RequestMapping("/resynchronize/{id}")
    public void resynchronize(@PathVariable("id") Long id){
        SErrorLog sErrorLog = isErrorLogService.selectSErrorLogById(id);
        try{
            sErrorLog.setStatus(2);
            isErrorLogService.updateSErrorLog(sErrorLog);
            RestHighLevelClient restHighLevelClient = esMap.get(sErrorLog.getConfigId());
            //连接查询结果
            Connection conn = mMap.get(sErrorLog.getConfigId());
            // 创建Statement对象
            String sql = "select * from "+sErrorLog.getTableName()+" where "+sErrorLog.getpKey()+"="+sErrorLog.getpKeyData();
            Statement stmt = conn.createStatement();
            ResultSet rs = stmt.executeQuery(sql);
            ResultSetMetaData metaData = rs.getMetaData();
            int columnCount = metaData.getColumnCount();
            // 处理查询结果
            List<Serializable> list = new ArrayList<>();
            while (rs.next()) {
                for(int i = 1; i <= columnCount; i++) {
                    Object value = rs.getObject(i);
                    list.add((Serializable)value);
                }
            }
            //同步到ES
            String dataObject = getDataObject(sErrorLog.getSyncType(),sErrorLog.getConfigId(),list,sErrorLog.getTableName());
            if(dataObject != null){
                log.info("ReSynchronize --> "+dataObject);
            }
            //同步成功
            sErrorLog.setStatus(3);
            isErrorLogService.updateSErrorLog(sErrorLog);
        }catch (Exception e){
            //同步失败
            sErrorLog.setStatus(1);
            sErrorLog.setErrorMsg(e.getMessage());
            sErrorLog.setCreateTime(new Date());
            isErrorLogService.updateSErrorLog(sErrorLog);
        }
    }

    /**
     * 创建监听器
     * @param sSyncConfig
     */
    public void create(SSyncConfig sSyncConfig){
        BinaryLogClient client = null;
        if(sSyncConfig.getmSchema() == null || "".equals(sSyncConfig.getmSchema())){
            client = new BinaryLogClient(sSyncConfig.getmUrl(), sSyncConfig.getmPort(),sSyncConfig.getmUser(), sSyncConfig.getmPass());
        }else{
            client = new BinaryLogClient(sSyncConfig.getmUrl(), sSyncConfig.getmPort(),sSyncConfig.getmSchema(),sSyncConfig.getmUser(), sSyncConfig.getmPass());
        }
        client.setHeartbeatInterval(30);
        client.setServerId(1);
        //监听处理
        client.registerEventListener(event -> {
            EventData data = event.getData();
            if (data instanceof TableMapEventData) {
                TableMapEventData tableMapEventData = (TableMapEventData) data;
                tableNameMap.put(tableMapEventData.getTableId(),tableMapEventData.getTable());
            }
            if (data instanceof UpdateRowsEventData) {
                if(sSyncConfig.getFilterate().indexOf("update") != -1){
                    UpdateRowsEventData  updateRowsEventData = (UpdateRowsEventData ) data;
                    for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
                        List<Serializable> entries = Arrays.asList(row.getValue());
                        String dataObject = getDataObject("update",sSyncConfig.getId(),entries,tableNameMap.get(updateRowsEventData.getTableId()));
                        if(dataObject != null){
                            log.info("UPDATE --> "+dataObject);
                        }
                    }
                }
            } else if (data instanceof WriteRowsEventData) {
                if(sSyncConfig.getFilterate().indexOf("insert") != -1){
                    WriteRowsEventData writeRowsEventData =(WriteRowsEventData) data;
                    List<Serializable[]> rows = writeRowsEventData.getRows();
                    for (Serializable[] row : rows) {
                        List<Serializable> entries = Arrays.asList(row);
                        String dataObject = getDataObject("insert",sSyncConfig.getId(),entries,tableNameMap.get(writeRowsEventData.getTableId()));
                        if(dataObject != null){
                            log.info("INSERT --> "+dataObject);
                        }
                    }
                }
            } else if (data instanceof DeleteRowsEventData) {
                if(sSyncConfig.getFilterate().indexOf("delete") != -1){
                    DeleteRowsEventData deleteRowsEventData =(DeleteRowsEventData) data;
                    List<Serializable[]> rows = deleteRowsEventData.getRows();
                    for (Serializable[] row : rows) {
                        List<Serializable> entries = Arrays.asList(row);
                        String dataObject = getDataObject("delete",sSyncConfig.getId(),entries,tableNameMap.get(deleteRowsEventData.getTableId()));
                        if(dataObject != null){
                            log.info("DELETE --> "+dataObject);
                        }
                    }
                }
            }
        });

        try {
            //数据库连接设置超时
            client.setConnectTimeout(30000);
            //监听 binlog
            mysqlMap.put(sSyncConfig.getId(),client);
            //ES 连接
            esMap.put(sSyncConfig.getId(),getEsClient(sSyncConfig.getId()));
            //MySQL 连接
            getMySqlClient(sSyncConfig.getId());
            mysqlMap.get(sSyncConfig.getId()).connect();
        } catch (Exception e) {
            mysqlMap.remove(sSyncConfig.getId());
            esMap.remove(sSyncConfig.getId());
            sSyncConfigMap.remove(sSyncConfig.getId());
            mMap.remove(sSyncConfig.getId());
            log.error("任务启动失败",e);
        }
    }

    /**
     * binlog 日志解析
     * @param message
     * @param tableName
     * @return
     */
    private String getDataObject(String type,Long id,List message,String tableName) {
        SSyncConfig sSyncConfig = sSyncConfigMap.get(id);
        JSONArray jsonArray = JSONArray.parseArray(sSyncConfig.getSyncData());
        String[] fieldArr = null;
        String[] mfieldArr = null;
        String[] efieldArr = null;
        String sql = null;
        String indexName = null;
        String pKey = null;
        if(jsonArray.size() !=0){
            Map<String,Object> resultObject = new HashMap<>();
            JSONObject json = new JSONObject();
            //字段处理
            for (int i = 0; i < jsonArray.size(); i++) {
                if(jsonArray.getJSONObject(i).getString("tableName").equals(tableName)){
                    indexName = jsonArray.getJSONObject(i).getString("indexName");
                    pKey = jsonArray.getJSONObject(i).getString("pKey");
                    //字段处理
                    fieldArr = jsonArray.getJSONObject(i).getString("field").split(",");
                    for (int j = 0; j < fieldArr.length; j++) {
                        json.put(fieldArr[j],j);
                    }
                    if("2".equals(sSyncConfig.getMappingType())){
                        mfieldArr = jsonArray.getJSONObject(i).getString("mfield").split(",");
                        efieldArr = jsonArray.getJSONObject(i).getString("efield").split(",");
                    }else if("3".equals(sSyncConfig.getMappingType())){
                        mfieldArr = jsonArray.getJSONObject(i).getString("mfield").split(",");
                        efieldArr = jsonArray.getJSONObject(i).getString("efield").split(",");
                        sql = jsonArray.getJSONObject(i).getString("sql");
                    }
                    break;
                }
            }
            for (String key : json.keySet()) {
                int i = Integer.parseInt(json.get(key).toString());
                //全表映射
                if(message.get(i) instanceof byte[] && !(message.get(i) instanceof String)){
                    try{
                        String str = new String((byte[]) message.get(i), "UTF-8");
                        resultObject.put(key, str);
                    }catch (Exception e){
                        resultObject.put(key, null);
                    }
                }else{
                    resultObject.put(key, message.get(i));
                }
            }

            try{
                //判断同步类型
                if("2".equals(sSyncConfig.getMappingType())){
                    //得到需要映射表
                    Map<String,Object> resultMap = new HashMap<>();
                    for (int j = 0; j < mfieldArr.length; j++) {
                        resultMap.put(efieldArr[j],resultObject.get(mfieldArr[j]));
                    }
                    resultObject = resultMap;
                }else if("3".equals(sSyncConfig.getMappingType())){
                    //查询语句条件替换
                    for (String key : resultObject.keySet()){
                        if(sql.indexOf("#"+key+"#") != -1){
                            sql = sql.replaceAll("#"+key+"#",resultObject.get(key).toString());
                        }
                    }
                    //连接查询结果
                    Connection conn = mMap.get(id);
                    // 创建Statement对象
                    Statement stmt = conn.createStatement();
                    ResultSet rs = stmt.executeQuery(sql);
                    // 处理查询结果
                    Map<String,Object> sqlResultMap = new HashMap<>();
                    while (rs.next()) {
                        for (int i = 0; i < mfieldArr.length; i++) {
                            sqlResultMap.put(mfieldArr[i],rs.getObject(mfieldArr[i]));
                        }
                    }
                    //得到需要映射表
                    Map<String,Object> resultMap = new HashMap<>();
                    for (int j = 0; j < mfieldArr.length; j++) {
                        resultMap.put(efieldArr[j],sqlResultMap.get(mfieldArr[j]));
                    }
                    resultObject = resultMap;
                }
                if(json.size() > 0){
                    RestHighLevelClient restHighLevelClient = esMap.get(id);
                    //同步到ES
                    syncEsData(restHighLevelClient,type,indexName,resultObject,pKey);
                    return JSONObject.toJSONString(resultObject);
                }
            }catch (Exception e){
                //记录 同步失败数据
                //序号、任务ID、失败时间、同步类型、同步数据、同步索引、主键
                //1、1、2023-01-01 12:12:10、insert、{}
                SErrorLog sErrorLog = new SErrorLog();
                sErrorLog.setConfigId(id);
                sErrorLog.setSyncType(type);
                sErrorLog.setTableName(tableName);
                sErrorLog.setpKey(pKey);
                sErrorLog.setSyncData(message.toString());
                sErrorLog.setpKeyData(resultObject.get(pKey).toString());
                sErrorLog.setErrorMsg(e.getMessage());
                sErrorLog.setStatus(1);
                isErrorLogService.insertSErrorLog(sErrorLog);
                return null;
            }
        }
        return null;
    }

    /**
     * 同步到ES
     * @param restHighLevelClient
     * @param indexName
     * @param resultObject
     */
    private void syncEsData(RestHighLevelClient restHighLevelClient,String type ,
                                   String indexName,Map<String, Object> resultObject,String pKey) throws Exception {
        Map<String, Object> jsonMap = new HashMap<>();
        for (Map.Entry<String, Object> entry : resultObject.entrySet()) {
            Object key = entry.getKey();
            Object value = entry.getValue();
            if(value != null){
                try{
                    JSONObject jsonObject = JSON.parseObject(String.valueOf(value));
                    Map<String,Object> maps = JSONObject.parseObject(jsonObject.toString(),Map.class);  ;
                    jsonMap.put(String.valueOf(key), maps);
                }catch (Exception e){
                    jsonMap.put(String.valueOf(key), value);
                }
            }
        }
        String id;
        if(pKey == null || "".equals(pKey)){
            id = String.valueOf(jsonMap.get("id"));
        }else{
            id = String.valueOf(jsonMap.get(pKey));
        }
        if("insert".equals(type)){
            IndexRequest indexRequest = new IndexRequest(indexName).id(id).source(jsonMap);
            restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        }else if("update".equals(type)){
            UpdateRequest updateRequest = new UpdateRequest(indexName, id);
            updateRequest.doc(jsonMap);
            restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
        }else if("delete".equals(type)){
            DeleteRequest deleteRequest = new DeleteRequest(indexName,id);
            restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
        }
    }

    /**
     * 创建ES 创建连接
     * @return
     */
    public RestHighLevelClient getEsClient(Long id) {
        SSyncConfig sSyncConfig = sSyncConfigMap.get(id);
        RestHighLevelClient client = null;
        if (client == null) {
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            if(sSyncConfig.geteUser() == null || sSyncConfig.getePass() == null){
                credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("", ""));
            }else{
                credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(sSyncConfig.geteUser(), sSyncConfig.getePass()));
            }
            client = new RestHighLevelClient(
                RestClient.builder(new HttpHost(sSyncConfig.geteUrl(), sSyncConfig.getePort(), sSyncConfig.geteHost()))
                    .setRequestConfigCallback(requestConfigBuilder -> {
                            return requestConfigBuilder
                                .setMaxRedirects(3) //最大重试次数
                                .setConnectTimeout(10000) //连接超时（默认为1秒）
                                .setSocketTimeout(60000);//套接字超时（默认为30秒）
                        }
                    ).setHttpClientConfigCallback(httpClientBuilder -> {
                            return httpClientBuilder
                                .setDefaultCredentialsProvider(credentialsProvider)
                                .setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build());//线程数
                        }
                    ));
        }
        return client;
    }

    /**
     * 创建数据库连接
     * @param id
     * @throws Exception
     */
    public void getMySqlClient(Long id) throws Exception {
        SSyncConfig sSyncConfig = sSyncConfigMap.get(id);
        // 加载JDBC驱动程序
        Class.forName("com.mysql.jdbc.Driver");
        // 建立数据库连接
        String url = "jdbc:mysql://"+sSyncConfig.getmUrl()+":"+sSyncConfig.getmPort()+"/"+sSyncConfig.getmSchema()+"?autoReconnect=true&failOverReadOnly=false&maxReconnects=10";
        String user = sSyncConfig.getmUser();
        String password = sSyncConfig.getmPass();
        Connection conn = DriverManager.getConnection(url, user, password);
        //添加到 MAP 里面去
        mMap.put(id,conn);
        //心跳异步任务
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                Long mId = id;
                while (true) {
                    try {
                        Statement stmt = mMap.get(mId).createStatement();
                        stmt.executeQuery("SELECT 1");
                        Thread.sleep(30000);
                        log.info("数据库心跳模式 --> * <--");
                    } catch (Exception e) {}
                }
            }
        });
        thread.start();
    }

}
